package xyz.sicau.test1;

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.ConfirmCallback;
import xyz.sicau.utils.MQUtils;

import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.util.concurrent.ConcurrentSkipListMap;
import java.util.concurrent.TimeoutException;

public class Producer {
    public static final String QUEUE_NAME = "hello";

    public static void main(String[] args) throws IOException, TimeoutException, InterruptedException {
        Channel channel = MQUtils.getChannel();

        channel.queueDeclare(QUEUE_NAME, true, false, false, null); // 这时已经有队列了，但是没有消息
        channel.confirmSelect();

        // 异步发布确认，最好的选择，用跳表来装当前已发送的消息，后续再不断的删除已确认的消息
        ConcurrentSkipListMap<Long, String> outstandingConfirms = new ConcurrentSkipListMap<>();
        ConfirmCallback ackCallback = (deliveryTag, multiple) -> { // 发布成功回调
            // headMap找出小于这个编号的键值对，并删除，不会删除编号对应的对，参数二为true时小于等于
            if (multiple) outstandingConfirms.headMap(deliveryTag).clear();
            outstandingConfirms.remove(deliveryTag);
        };
        ConfirmCallback nackCallback = (deliveryTag, multiple) -> { // 发布失败回调
            System.out.println(deliveryTag + "发布失败：" + outstandingConfirms.get(deliveryTag));
        };
        channel.addConfirmListener(ackCallback, nackCallback); // 消息发布监听器

        for (int i = 0; i < 100; i++) { // 只管发就完了
            String msg = i + "消息";
            // 发了的消息记录在表中，channel.getNextPublishSeqNo()第一次访问就是1，从1开始！！！
            outstandingConfirms.put(channel.getNextPublishSeqNo(), msg);
            channel.basicPublish("", QUEUE_NAME, null, msg.getBytes(StandardCharsets.UTF_8));
        }

    }
}
